/*
 * Copyright (C) 2011 The Guava Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.ttzero.excel.hash;


import java.io.Serializable;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.function.Predicate;

/**
 * A Bloom filter for instances of {@code String}. A Bloom filter offers an approximate containment test
 * with one-sided error: if it claims that an element is contained in it, this might be in error,
 * but if it claims that an element is <i>not</i> contained in it, then this is definitely true.
 *
 * <p>If you are unfamiliar with Bloom filters, this nice <a
 * href="http://llimllib.github.com/bloomfilter-tutorial/">tutorial</a> may help you understand how
 * they work.
 *
 * <p>The false positive probability ({@code FPP}) of a Bloom filter is defined as the probability
 * that {@linkplain #mightContain(String)} will erroneously return {@code true} for an object that
 * has not actually been put in the {@code BloomFilter}.
 *
 * <p>Bloom filters are serializable. Both serialized forms will continue to be
 * supported by future versions of this library. However, serial forms generated by newer versions
 * of the code may not be readable by older versions of the code (e.g., a serialized Bloom filter
 * generated today may <i>not</i> be readable by a binary that was compiled 6 months ago).
 *
 * <p>As of Guava 23.0, this class is thread-safe and lock-free. It internally uses atomics and
 * compare-and-swap to ensure correctness when multiple threads are used to access it.
 *
 * @author Dimitris Andreou
 * @author Kevin Bourrillion
 * @since 11.0 (thread-safe since 23.0)
 */
public final class StringBloomFilter implements Predicate<String>, Serializable {
    @Override
    public boolean test(String t) {
        return mightContain(t);
    }

    /**
     * A strategy to translate T instances, to {@code numHashFunctions} bit indexes.
     *
     * <p>Implementations should be collections of pure functions (i.e. stateless).
     */
    static class Strategy {
        // Hasher Functions
        private final Murmur3_128Hasher hasher = new Murmur3_128Hasher(0);
        /**
         * Sets {@code numHashFunctions} bits of the given bit array, by hashing a user element.
         *
         * <p>Returns whether any bits changed as a result of this operation.
         */
        public boolean put(String object, Charset charset, int numHashFunctions, LockFreeBitArray bits) {
            long bitSize = bits.bitSize();
            byte[] bytes = hasher.clear().putBytes(object.getBytes(charset)).hash();
            long hash1 = fromBytes(bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]);
            long hash2 = fromBytes(bytes[15], bytes[14], bytes[13], bytes[12], bytes[11], bytes[10], bytes[9], bytes[8]);

            boolean bitsChanged = false;
            long combinedHash = hash1;
            for (int i = 0; i < numHashFunctions; i++) {
                // Make the combined hash positive and indexable
                bitsChanged |= bits.set((combinedHash & Long.MAX_VALUE) % bitSize);
                combinedHash += hash2;
            }
            return bitsChanged;
        }

        /**
         * Queries {@code numHashFunctions} bits of the given bit array, by hashing a user element;
         * returns {@code true} if and only if all selected bits are set.
         */
        public boolean mightContain(String object, Charset charset, int numHashFunctions, LockFreeBitArray bits) {
            long bitSize = bits.bitSize();
            byte[] bytes = hasher.clear().putBytes(object.getBytes(charset)).hash();
            long hash1 = fromBytes(bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]);
            long hash2 = fromBytes(bytes[15], bytes[14], bytes[13], bytes[12], bytes[11], bytes[10], bytes[9], bytes[8]);

            long combinedHash = hash1;
            for (int i = 0; i < numHashFunctions; i++) {
                // Make the combined hash positive and indexable
                if (!bits.get((combinedHash & Long.MAX_VALUE) % bitSize)) {
                    return false;
                }
                combinedHash += hash2;
            }
            return true;
        }

        /**
         * Returns the {@code long} value whose byte representation is the given 8 bytes, in big-endian
         * order; equivalent to {@code Longs.fromByteArray(new byte[] {b1, b2, b3, b4, b5, b6, b7, b8})}.
         *
         * @since 7.0
         */
        public static long fromBytes(byte b1, byte b2, byte b3, byte b4, byte b5, byte b6, byte b7, byte b8) {
            return (b1 & 0xFFL) << 56
                    | (b2 & 0xFFL) << 48
                    | (b3 & 0xFFL) << 40
                    | (b4 & 0xFFL) << 32
                    | (b5 & 0xFFL) << 24
                    | (b6 & 0xFFL) << 16
                    | (b7 & 0xFFL) << 8
                    | (b8 & 0xFFL);
        }
    }

    /**
     * The bit set of the BloomFilter (not necessarily power of 2!)
     */
    private final LockFreeBitArray bits;

    /**
     * Number of hashes per element
     */
    private final int numHashFunctions;

    /**
     * The funnel to translate Ts to bytes
     */
    private final Charset charset;

    /**
     * The strategy we employ to map an element T to {@code numHashFunctions} bit indexes.
     */
    private final Strategy strategy;

    /**
     * Creates a BloomFilter.
     */
    private StringBloomFilter(LockFreeBitArray bits, int numHashFunctions, Charset charset, Strategy strategy) {
        this.bits = bits;
        this.numHashFunctions = numHashFunctions;
        this.charset = charset;
        this.strategy = strategy;
    }

    /**
     * Creates a new {@code BloomFilter} that's a copy of this instance. The new instance is equal to
     * this instance but shares no mutable state.
     *
     * @return a {@link StringBloomFilter} copy from current filter
     * @since 12.0
     */
    public StringBloomFilter copy() {
        return new StringBloomFilter(bits.copy(), numHashFunctions, charset, strategy);
    }

    /**
     * Returns {@code true} if the element <i>might</i> have been put in this Bloom filter, {@code
     * false} if this is <i>definitely</i> not the case.
     */
    public boolean mightContain(String object) {
        return strategy.mightContain(object, charset, numHashFunctions, bits);
    }

    /**
     * Puts an element into this {@code BloomFilter}. Ensures that subsequent invocations of {@link
     * #mightContain(String)} with the same element will always return {@code true}.
     *
     * @return true if the Bloom filter's bits changed as a result of this operation. If the bits
     * changed, this is <i>definitely</i> the first time {@code object} has been added to the
     * filter. If the bits haven't changed, this <i>might</i> be the first time {@code object} has
     * been added to the filter. Note that {@code put(t)} always returns the <i>opposite</i>
     * result to what {@code mightContain(t)} would have returned at the time it is called.
     * @since 12.0 (present in 11.0 with {@code void} return type})
     */
    public boolean put(String object) {
        return strategy.put(object, charset, numHashFunctions, bits);
    }

    /**
     * Combines this Bloom filter with another Bloom filter by performing a bitwise OR of the
     * underlying data. The mutations happen to <b>this</b> instance. Callers must ensure the Bloom
     * filters are appropriately sized to avoid saturating them.
     *
     * @param that The Bloom filter to combine this Bloom filter with. It is not mutated.
     * @throws IllegalArgumentException if {@code isCompatible(that) == false}
     * @since 15.0
     */
    public void putAll(StringBloomFilter that) {
        this.bits.putAll(that.bits);
    }

    @Override
    public boolean equals(Object object) {
        if (object == this) {
            return true;
        }
        if (object instanceof StringBloomFilter) {
            StringBloomFilter that = (StringBloomFilter) object;
            return this.numHashFunctions == that.numHashFunctions
                    && this.charset.equals(that.charset)
                    && this.bits.equals(that.bits)
                    && this.strategy.equals(that.strategy);
        }
        return false;
    }

    @Override
    public int hashCode() {
        return Arrays.hashCode(new Object[]{numHashFunctions, charset, strategy, bits});
    }

    public static StringBloomFilter create(long expectedInsertions, double fpp) {
        return create(StandardCharsets.UTF_8, expectedInsertions, fpp);
    }

    /**
     * Creates a {@link StringBloomFilter} with the expected number of insertions and expected false
     * positive probability.
     *
     * <p>Note that overflowing a {@code BloomFilter} with significantly more elements than specified,
     * will result in its saturation, and a sharp deterioration of its false positive probability.
     *
     * <p>The constructed {@code BloomFilter} will be serializable if the provided {@code Funnel<T>}
     * is.
     *
     * <p>It is recommended that the funnel be implemented as a Java enum. This has the benefit of
     * ensuring proper serialization and deserialization, which is important since {@link #equals}
     * also relies on object identity of funnels.
     *
     * @param charset            String funnel charset
     * @param expectedInsertions the number of expected insertions to the constructed {@code
     *                           BloomFilter}; must be positive
     * @param fpp                the desired false positive probability (must be positive and less than 1.0)
     * @return a {@code BloomFilter}
     * @since 19.0
     */
    public static StringBloomFilter create(Charset charset, long expectedInsertions, double fpp) {

        if (expectedInsertions == 0) {
            expectedInsertions = 1;
        }
        /*
         * is proportional to -log(p), but there is not much of a point after all, e.g.
         * optimalM(1000, 0.0000000000000001) = 76680 which is less than 10kb. Who cares!
         */
        long numBits = optimalNumOfBits(expectedInsertions, fpp);
        int numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, numBits);
        try {
            return new StringBloomFilter(new LockFreeBitArray(numBits), numHashFunctions, charset, new Strategy());
        } catch (IllegalArgumentException e) {
            throw new IllegalArgumentException("Could not create BloomFilter of " + numBits + " bits", e);
        }
    }

    // Cheat sheet:
    //
    // m: total bits
    // n: expected insertions
    // b: m/n, bits per insertion
    // p: expected false positive probability
    //
    // 1) Optimal k = b * ln2
    // 2) p = (1 - e ^ (-kn/m))^k
    // 3) For optimal k: p = 2 ^ (-k) ~= 0.6185^b
    // 4) For optimal k: m = -nlnp / ((ln2) ^ 2)

    /**
     * Computes the optimal k (number of hashes per element inserted in Bloom filter), given the
     * expected insertions and total number of bits in the Bloom filter.
     *
     * <p>See http://en.wikipedia.org/wiki/File:Bloom_filter_fp_probability.svg for the formula.
     *
     * @param n expected insertions (must be positive)
     * @param m total number of bits in Bloom filter (must be positive)
     */
    static int optimalNumOfHashFunctions(long n, long m) {
        // (m / n) * log(2), but avoid truncation due to division!
        return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
    }

    /**
     * Computes m (total bits of Bloom filter) which is expected to achieve, for the specified
     * expected insertions, the required false positive probability.
     *
     * <p>See http://en.wikipedia.org/wiki/Bloom_filter#Probability_of_false_positives for the
     * formula.
     *
     * @param n expected insertions (must be positive)
     * @param p false positive rate (must be 0 < p < 1)
     */
    static long optimalNumOfBits(long n, double p) {
        if (p == 0) {
            p = Double.MIN_VALUE;
        }
        return (long) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));
    }
}
